package belf.migrate.engine.job.verify;

import belf.migrate.api.job.Job;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.util.PostLog;
import belf.migrate.api.util.VerifyObjectTypes;
import belf.migrate.engine.BelfEvn;
import belf.migrate.engine.job.*;
import belf.migrate.engine.job.model.verify.*;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;

import java.text.SimpleDateFormat;
import java.util.Date;

import static belf.migrate.api.job.JobCode.*;

/**
 * 迁移后对迁移过程的结果进行验证，如： 数据表行数、最新数据的主键 对数据行进行抽查
 */
@Slf4j
public class VerifyJobRunner extends BaseJobRunner {

    private final SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    private static Gson gson = new Gson();

    public VerifyJobRunner(Job job) {
        super(job);
    }


    public JobResult check() {
        JobResult jobResult = new JobResult(SUCCESS, "任务执行成功");
        jobResult.setStartTime(simpleDateFormat.format(new Date()));

        jobContext = new JobContext("" + job.getJobId(), taskConf);
        // 创建源端catalog
        AbstractJdbcCatalog SourceCatalog = CatalogCreator.createSourceCatalog(taskConf);
        if (null == SourceCatalog) {
            log.error("No Source Catalog for {}, Stop Job {}!",
                    taskConf.getSourceConf().getCatalogName(),
                    job.getJobId());

            jobResult = new VerifyJobResult(NO_SOURCE_CATALOG, "提交Job失败，没有找到目标端" +
                    taskConf.getSinkConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");

            return jobResult;
        }
        jobContext.setSourceCatalog(SourceCatalog);

        // 创建目标catalog
        AbstractSinkCatalog sinkCatalog = CatalogCreator.createSinkCatalog(taskConf);
        if (null == sinkCatalog) {
            log.error("No Sink Catalog for {}, Stop Job {}!",
                    taskConf.getSinkConf().getCatalogName(),
                    job.getJobId());

            jobResult = new VerifyJobResult(NO_SINK_CATALOG, "提交Job失败，没有找到源端" +
                    taskConf.getSourceConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");
            return jobResult;
        }
        jobContext.setSinkCatalog(sinkCatalog);
        BelfEvn.getInstance().addJobContext("" + job.getJobId(), jobContext);

        switch (job.getTask().getTaskType()) {
            case VERIFY_TABLE:
            case VERIFY_VIEW:
            case VERIFY_TRIGGER:
            case VERIFY_SEQ:
            case VERIFY_SYNONYM:
            case VERIFY_FUNC:
            case VERIFY_TYPE:
                jobContext.setTotalProgress(1);
                break;
            case VERIFY_MIGRATION:
                jobContext.setTotalProgress(VerifyObjectTypes.getObjectTypeSize(job.getTask().getTaskConf().getSourceConf().getCatalogName(),
                        job.getTask().getTaskConf().getSinkConf().getCatalogName()));
                break;


        }

        return jobResult;
    }

    @Override
    public JobResult execute() {
        log.info("校验开始！");

        JobResult jobResult = check();
        if (jobResult.getCode() != SUCCESS) {
            return jobResult;
        }
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "校验开始！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        VerifyAllResult verifyAllResult = new VerifyAllResult();
        long startTime = System.currentTimeMillis();
        switch (job.getTask().getTaskType()) {

            case VERIFY_TABLE:
                VerifyTableResult tableResult = new VerifyTableTool(jobContext, job).run();
                verifyAllResult.setTableResult(tableResult);
                if (FAILTURE == tableResult.getCode()) {
                    jobResult.setCode(FAILTURE);
                }
                break;
            case VERIFY_VIEW:
                VerifyViewResult viewResult = new VerifyViewTool(jobContext, job).run();
                verifyAllResult.setViewResult(viewResult);
                if (FAILTURE == viewResult.getCode()) {
                    jobResult.setCode(FAILTURE);
                }
                break;
            case VERIFY_TRIGGER:
                VerifyTriggerResult triggerResult = new VerifyTriggerTool(jobContext, job).run();
                verifyAllResult.setTriggerResult(triggerResult);
                if (FAILTURE == triggerResult.getCode()) {
                    jobResult.setCode(FAILTURE);
                }
                break;
//            case VERIFY_INDEX:
//                VerifyIndexResult indexResult = new VerifyIndexTool(jobContext, job).run();
//                verifyAllResult.setIndexResult(indexResult);
//                if (FAILTURE == indexResult.getCode()) {
//                    jobResult.setCode(FAILTURE);
//                }
//                break;
            case VERIFY_SEQ:
                VerifySeqResult seqResult = new VerifySeqTool(jobContext, job).run();
                verifyAllResult.setSeqResult(seqResult);
                if (FAILTURE == seqResult.getCode()) {
                    jobResult.setCode(FAILTURE);
                }
                break;
            case VERIFY_SYNONYM:
                VerifySynonymResult synonymResult = new VerifySynonymTool(jobContext, job).run();
                verifyAllResult.setSynonymResult(synonymResult);
                if (FAILTURE == synonymResult.getCode()) {
                    jobResult.setCode(FAILTURE);
                }
                break;
            case VERIFY_FUNC:
                VerifyFunctionResult functionResult = new VerifyFuncTool(jobContext, job).run();
                verifyAllResult.setFuncResult(functionResult);
                if (FAILTURE == functionResult.getCode()) {
                    jobResult.setCode(FAILTURE);
                }
                break;
            case VERIFY_TYPE:
                VerifyTypeResult typeResult = new VerifyTypeTool(jobContext, job).run();
                verifyAllResult.setTypeResult(typeResult);
                if (FAILTURE == typeResult.getCode()) {
                    jobResult.setCode(FAILTURE);
                }
                break;
            case VERIFY_MIGRATION:
                verifyAllResult = new VerifyAllTool(
                        new VerifyTableTool(jobContext, job),
                        new VerifyViewTool(jobContext, job),
                        new VerifyTriggerTool(jobContext, job),
                        new VerifySeqTool(jobContext, job),
                        new VerifySynonymTool(jobContext, job),
                        new VerifyFuncTool(jobContext, job),
                        new VerifyTypeTool(jobContext, job)).run();
                if (FAILTURE == verifyAllResult.getCode()) {
                    jobResult.setCode(FAILTURE);
                }
                break;
            default:
                break;

        }
        // 计算耗时
        long endTime = System.currentTimeMillis();
        String timeConsuming = String.valueOf((double) (endTime - startTime) / 1000);
        verifyAllResult.setAllTimeConsuming(String.format("%s秒", timeConsuming));
        jobResult.setData(verifyAllResult);
        jobResult.setJobId(job.getJobId());
        log.info("校验结束！");
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "校验结束！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());

        jobResult.setEndTime(simpleDateFormat.format(new Date()));
        jobResult.setTimeConsuming(timeConsuming);
        return jobResult;
    }
}
